package com.atguigu.mapjoin;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class MJMapper extends Mapper<LongWritable, Text,OrderBean, NullWritable> {
    //封装key
    private OrderBean orderBean = new OrderBean();
    //用来缓存pd.txt中的内容（key为pid,value为pname）
    private Map<Integer,String> map = new HashMap<Integer,String>();
    /*
        在任务开始的时候执行一次。在map方法前执行
        作用 ：将pd.txt中的内容缓存到内存中（缓存到Map中）
     */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        FileSystem fs = null;
        FSDataInputStream fis = null;
        BufferedReader br = null;
        try {
            //1.创建文件系统对象
            fs = FileSystem.get(context.getConfiguration());
            //2.创建流
            //2.1获取缓存文件的路径
            URI[] cacheFiles = context.getCacheFiles();
            //2.2创建流
            fis = fs.open(new Path(cacheFiles[0]));
            //需要（一行一行的读取内容）
            br = new BufferedReader(new InputStreamReader(fis,"utf-8"));

            //3.读取数据
            String line = "";
            while ((line = br.readLine()) != null) {
                //4.切割数据
                String[] split = line.split("\t");
                //5.将数据存放到map中
                map.put(Integer.parseInt(split[0]), split[1]);
            }
        }catch (Exception e){
            e.printStackTrace();
            throw new RuntimeException(e.getMessage());//终止程序的运行
        }finally {
            //6.关闭资源
            if (br != null){
                br.close();
            }
            if (fis != null){
                fis.close();
            }
            if (fs != null){
                fs.close();
            }
        }


    }

    /*
        作用 ：1.读取order.txt中的内容    2.进行join的操作
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1.将Text转成String
        String line = value.toString();
        //2.切割数据
        String[] split = line.split("\t");
        //3.封装key
        orderBean.setId(Integer.parseInt(split[0]));
        orderBean.setPid(Integer.parseInt(split[1]));
        orderBean.setAmount(Integer.parseInt(split[2]));
        orderBean.setPname(map.get(Integer.parseInt(split[1])));
        //4.写出key,value
        context.write(orderBean,NullWritable.get());
    }

    /*
        关闭资源
     */
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {

    }
}
